package com.demo.function.thread;

import lombok.extern.slf4j.Slf4j;

import java.util.ArrayList;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;

/**
 * 使用Java8 多线程CompletableFuture
 */
@Slf4j
public class Jdk8Thread {

    public void completableFuture() throws ExecutionException, InterruptedException {
        long start = System.currentTimeMillis();
        List<CompletableFuture<Map<String, Integer>>> list = new ArrayList<>();

        CountDownLatch countDownLatch = new CountDownLatch(1);

        for (int i = 0; i < 1000; i++) {
            final int tmp = i;
            Map<String, Integer> map = new HashMap<>();
            // CompletableFuture不传线程池，默认创建线程池，size=3
            CompletableFuture<Map<String, Integer>> mapCompletableFuture = CompletableFuture.supplyAsync(() -> {
                try {
                    countDownLatch.await();
                    log.info("i = {}", tmp);

                    map.put("i", tmp);
                    map.put("random", (int)(Math.random() * 1000));

                } catch (InterruptedException e) {
                    e.printStackTrace();
                } finally {
                    countDownLatch.countDown();
                }
                return map;
            });

            list.add(mapCompletableFuture);
        }

        log.info("countdown end");
        countDownLatch.countDown();

        for (CompletableFuture<Map<String, Integer>> mapCompletableFuture : list) {
            Map<String, Integer> stringIntegerMap = mapCompletableFuture.get();
            log.info("map:{}", stringIntegerMap);
        }
        log.info("testCompletableFuture耗时：{}", (System.currentTimeMillis() - start));
    }

    public void completableFutureExecutors() throws ExecutionException, InterruptedException {
        /*
         *  创建线程池，会更加所需的并行层次来动态创建和关闭线程。
         *  它同样会试图减少任务队列的大小，所以比较适于高负载的环境。
         *  同样也比较适用于当执行的任务会创建更多任务，如递归任务。
         *  适合使用在很耗时的操作，但是newWorkStealingPool不是ThreadPoolExecutor的扩展，它是新的线程池类ForkJoinPool的扩展，
         *  但是都是在统一的一个Executors类中实现，由于能够合理的使用CPU进行对任务操作（并行操作），所以适合使用在很耗时的任务中
         *
         */
        ExecutorService executorService = Executors.newWorkStealingPool(30);
        long start = System.currentTimeMillis();
        List<CompletableFuture<Map<String, Integer>>> list = new ArrayList<>();

        CountDownLatch countDownLatch = new CountDownLatch(1);

        for (int i = 0; i < 300; i++) {
            final int tmp = i;
            Map<String, Integer> map = new HashMap<>();
            CompletableFuture<Map<String, Integer>> mapCompletableFuture = CompletableFuture.supplyAsync(() -> {
                try {
                    countDownLatch.await();
                    log.info("i = {}", tmp);

                    map.put("i", tmp);
                    map.put("random", (int)(Math.random() * 1000));
                    Thread.sleep(5000);
                } catch (InterruptedException e) {
                    e.printStackTrace();
                } finally {
                    countDownLatch.countDown();
                }
                return map;
            }, executorService);

            list.add(mapCompletableFuture);
        }

        log.info("countdown end");
        countDownLatch.countDown();

        for (CompletableFuture<Map<String, Integer>> mapCompletableFuture : list) {
            Map<String, Integer> stringIntegerMap = mapCompletableFuture.get();
            log.info("map:{}", stringIntegerMap);
        }
        log.info("testCompletableFutureExecutors耗时：{}", (System.currentTimeMillis() - start));
    }
}
